package com.lizhiyu.flink.demo4_transformation;

import com.lizhiyu.flink.model.VideoOrder;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Date;
import java.util.Objects;


/***
 * Reduce  可以对一些分组的信息进行聚合操作,比max 等方法更灵活
 */
public class ReduceTransformMation {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<VideoOrder> ds = env.fromCollection(
                Arrays.asList(
                        new VideoOrder("1", "java ", 1, 2019,new Date()),
                        new VideoOrder("1", "linux ", 2, 2019,new Date()),
                        new VideoOrder("1", "db ", 3, 2019,new Date()),
                        new VideoOrder("1", "java ", 1, 2019,new Date()),
                        new VideoOrder("1", "linux ", 2, 2019,new Date()),
                        new VideoOrder("1", "linux ", 2, 2019,new Date()),
                        new VideoOrder("1", "db ", 3, 2019,new Date())
                )
        );

        //过滤  <T, O>  T是参数类型 O是返回的参数类型
        SingleOutputStreamOperator<VideoOrder> filterDs = ds.flatMap(new FlatMapFunction<VideoOrder, VideoOrder>() {
            @Override
            public void flatMap(VideoOrder value, Collector<VideoOrder> out) throws Exception {
                if (Objects.nonNull(value) /*&& NumberUtils.INTEGER_ONE.equals(value.getMoney())*/) {
                    //去掉money不是1的数据
                    out.collect(value);
                }
            }
        });
        //看  keySelector 中的泛型为<IN, KEY>  IN 输入值类型 和  分组KEY的类型
        KeyedStream<VideoOrder, String> keyByDs = filterDs.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder value) throws Exception {
                return  value.getTitle();
            }
        });
        //求和的字段属性是  money
//        SingleOutputStreamOperator<VideoOrder> reduce = keyByDs.reduce(new ReduceFunction<VideoOrder>() {
//            @Override
//            public VideoOrder reduce(VideoOrder value1, VideoOrder value2) throws Exception {
//                System.out.println("--第一次数据时不会执行这里的-- 从第二条数据来了才会走这里  --value2 是上一次返回的对象  过了第一次每有一条数据则执行一次此方法");
//                //每个分后后的仅仅会对应一条数据，这条数据就存在value2 中
//                return VideoOrder.builder().money(value1.getMoney() + value2.getMoney()).title(value1.getTitle()).build();
//            }
//        });
//        reduce.print();

        SingleOutputStreamOperator<VideoOrder> reduce = keyByDs.process(new KeyedProcessFunction<String, VideoOrder, VideoOrder>() {
            @Override
            public void processElement(VideoOrder value, Context ctx, Collector<VideoOrder> out) throws Exception {

            }
        });
        env.execute("test");
    }
}
